package org.redis.cluster.loadbalance.slot;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.commons.collections.CollectionUtils;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.config.ClientConfig;
import org.redis.cluster.config.ClientInfo;
import org.redis.cluster.config.ClientStatus;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.registry.Listener;
import org.redis.cluster.registry.Registry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Jedis;

public class SlotReplicater implements Plugin,Listener {
	
	protected final Logger logger=LoggerFactory.getLogger(getClass());

	private RedisCluster redisCluster;
	private Registry registry;
	private SlotManager slotManager;
	
	private int timeout;
	private int thread;
	private ExecutorService executorService;
	
	public SlotReplicater(SlotManager slotManager) {
		this.slotManager=slotManager;
	}
	
	@Override
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.registry=redisCluster.getRegistry();
		this.timeout=redisCluster.getConfigManager().getSystemConfig().get(SlotConstants.SLOT_LEADER_TIMEOUT_SYSTEM_CONFIG).getValue(Integer.class, SlotConstants.DEFAULT_SLOT_LEADER_TIMEOUT);
		this.thread=redisCluster.getConfigManager().getSystemConfig().get(SlotConstants.SLOT_LEADER_THREAD_SYSTEM_CONFIG).getValue(Integer.class, SlotConstants.DEFAULT_SLOT_LEADER_THREAD);
		this.executorService=initExecutorService();
		initReplicateListener();
	}
	
	@Override
	public void destroy() {
	}
	
	protected ExecutorService initExecutorService() {
		return Executors.newFixedThreadPool(thread);
	}
	
	protected void initReplicateListener() {
		registry.listen(SlotHelper.getSlotPath(SlotConstants.REPLICATE_PATH), (String path)->{
			registry.selectLeader(SlotHelper.getSlotPath(SlotConstants.LEADER_PATH), timeout, this);
		});
	}

	@Override
	public void onListen(String path) {
		if(logger.isInfoEnabled()) {
			logger.info("处理槽数据复制");
		}
		
		Collection<SlotClientConfig> configs=slotManager.getSlotClientConfigs();
		if(CollectionUtils.isEmpty(configs)) {
			if(logger.isInfoEnabled()) {
				logger.info("没有需要处理的槽数据复制");
			}
			
			return;
		}
		
		if(logger.isInfoEnabled()) {
			logger.info("共有{}个需要处理的槽数据复制",configs.size());
		}
		
		for(SlotClientConfig slotClientConfig:configs) {
			ClientConfig clientConfig=SlotClientConfigAdapter.adaptClientConfig(slotClientConfig);
			if(logger.isInfoEnabled()) {
				logger.info("准备处理 {} 的数据复制",slotClientConfig);
			}
			
			if(!validateConfig(slotClientConfig)) {
				if(logger.isInfoEnabled()) {
					logger.info("验证失败 - {}",slotClientConfig);
				}
				
				clientConfig.setInfo(new ClientInfo(ClientStatus.ERROR));
				registry.registerClient(clientConfig);
			}
			else {
				clientConfig.setInfo(new ClientInfo(ClientStatus.REPLICATING));
				registry.registerClient(clientConfig);
				
				doReplicateProcess(slotClientConfig);
				
				clientConfig.setInfo(new ClientInfo(ClientStatus.NORMAL));
				registry.registerClient(clientConfig);
			}
			
			if(logger.isInfoEnabled()) {
				logger.info("完成 {} 的槽数据复制",slotClientConfig);
			}
			
			slotManager.unregisterSlotClifngConfig(slotClientConfig);
		}
	}
	
	protected boolean validateConfig(SlotClientConfig config) {
		try {
			Jedis jedis=new Jedis(config.getHost(),config.getPort(),config.getTimeout());
			jedis.ping();
			jedis.close();
			return true;
		}
		catch(Throwable t) {
			if(logger.isDebugEnabled()) {
				logger.debug("",t);
			}
			
			return false;
		}
	}
	
	protected void doReplicateProcess(SlotClientConfig config) {
		Collection<ClientConfig> clientConfigs=registry.getClients();
		Map<String,Collection<Integer>> slotMap=getSlotDataMap(config);
		
		Collection<ClientConfig> sources=new ArrayList<ClientConfig>(slotMap.size());
		for(String id:slotMap.keySet()) {
			for(ClientConfig tempClientConfig:clientConfigs) {
				if(!tempClientConfig.getId().equals(id) || (tempClientConfig.getHost().equals(config.getHost()) && tempClientConfig.getPort().equals(config.getPort()))) {
					continue;
				}
				
				sources.add(tempClientConfig);
				break;
			}
		}
		
		Map<String,Boolean> resultMap=new HashMap<String,Boolean>(sources.size());
		for(ClientConfig source:sources) {
			boolean result=false;
			try {
				result=doReplicateProcessor(source, config);
				if(logger.isInfoEnabled()) {
					logger.info("{} -> {} 的槽数据复制完成",source,config);
				}
			}
			catch(Throwable t) {
				if(logger.isInfoEnabled()) {
					logger.info("{} -> {} {} 的槽数据复制失败 - ",slotMap.get(source.getId()),source,config,t);
				}
			}
			
			resultMap.put(source.getId(), result);
		}

		if(CollectionUtils.isEmpty(sources)) {
			resetSlotData(config, null);
			
			if(logger.isInfoEnabled()) {
				logger.info("完成 {} 的槽数据重置",Objects.toString(config.getSlots()));
			}
		}
		else {
			for(ClientConfig source:sources) {
				if(resultMap.containsKey(source.getId()) && resultMap.get(source.getId()).equals(false)) {
					continue;
				}
				
				resetSlotData(config,source);
				
				try {
					doReplicateProcessor(source, config);
					if(logger.isInfoEnabled()) {
						logger.info("{} -> {} 的槽数据二次复制完成",source,config);
					}
				}
				catch(Throwable t) {
					if(logger.isInfoEnabled()) {
						logger.info("{} -> {} {} 的槽数据二次复制失败 - ",slotMap.get(source.getId()),source,config,t);
					}
				}
				
				if(logger.isInfoEnabled()) {
					logger.info("完成 {} -> {} 的槽数据重置",source.getId(),config.getId());
				}
			}
		}
	}
	
	protected void resetSlotData(SlotClientConfig config,ClientConfig source) {
		SlotMapping slotMapping=slotManager.getSlotMapping(config.getId());
		if(slotMapping==null) {
			slotMapping=new SlotMapping();
			slotMapping.setClientId(config.getId());
			slotMapping.setSlots(new ArrayList<Integer>());
		}
		
		for(int index:config.getSlots()) {
			slotMapping.getSlots().add(index);
			slotManager.setSlotClientId(index, config.getId());
		}
		
		slotManager.setSlotMapping(slotMapping);
		
		SlotMapping sourceSlotMapping=null;
		if(source!=null) {
			sourceSlotMapping=slotManager.getSlotMapping(source.getId());
			for(int index:config.getSlots()) {
				sourceSlotMapping.getSlots().remove((Object) index);
			}
			
			if(CollectionUtils.isEmpty(sourceSlotMapping.getSlots())) {
				registry.unregisterClient(source);
				slotManager.removeSlotMapping(source.getId());
			}
			else {
				slotManager.setSlotMapping(sourceSlotMapping);
			}
		}
	}
	
	protected Map<String,Collection<Integer>> getSlotDataMap(SlotClientConfig config) {
		Map<String,Collection<Integer>> result=new HashMap<String,Collection<Integer>>();
		
		for(int index:config.getSlots()) {
			String clientId=slotManager.getSlotClientId(index);
			Collection<Integer> slots=result.get(clientId);
			
			if(slots==null) {
				slots=new ArrayList<Integer>();
				result.put(clientId, slots);
			}
			
			slots.add(index);
		}
		
		return result;
	}
	
	protected boolean doReplicateProcessor(ClientConfig source,SlotClientConfig target) throws Throwable {
		Future<Boolean> future=executorService.submit(new SlotReplicateProcessor(source,target));
		future.get(timeout, TimeUnit.MILLISECONDS);
		return true;
	}
}